/*
实现异步工作器
 */
#ifndef __M__LOOPER_H__
#define __M__LOOPER_H__
#include "buffer.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>
#include <atomic>

namespace xglog
{
  using Functor = std::function<void(Buffer &)>; // 回调函数
  enum class AsyncType
  {
    ASYNC_SAFE,
    ASYNC_UNSAFE,
  };
  class AsyncLooper
  {
  public:
    using ptr = std::shared_ptr<AsyncLooper>;
    AsyncLooper(const Functor &cb, AsyncType loop_type = AsyncType::ASYNC_SAFE) : _stop(false), _thread(std::thread(&AsyncLooper::threadEntry, this)),
                                                                                  _callBack(cb), _looper_type(loop_type) {}
    ~AsyncLooper()
    {
      stop();
    }
    void stop()
    {
      _stop = true;
      _cond_con.notify_all(); // 停止日志器的工作，然后唤醒所有的消费者线程，通知他们停止工作。
      _thread.join();         // 等待工作线程的退出
    }
    void push(const char *data, size_t len)
    {
      // 1.无线扩容，非安全 //2.固定大小--生产缓冲区满了就阻塞
      std::unique_lock<std::mutex> lock(_mutex); // 加锁，防止多个线程同时往一个缓冲区放资源
      // 条件变量控制，若缓冲区剩余空间大小大于数据长度，则可以添加数据，生产者等。满足条件才能添加数据
      if (_looper_type == AsyncType::ASYNC_SAFE)
        _cond_pro.wait(lock, [&]()
                       { return _pro_buf.writeAbleSize() >= len; });
      // 满足条件，就可以添加数据
      _pro_buf.push(data, len);
      // 唤醒消费者缓冲区来处理
      _cond_con.notify_one(); // 唤醒一个消费者缓冲区来处理数据
    }

  private:
    Functor _callBack; // 具体对缓冲区数据进行处理的回调函数，由异步工作器提供
    // 就是先给_callBack先设置一个函数
    // 然后进行触发回调
  private:
    void threadEntry() // 线程入口函数,对消费者缓冲区的数据进行处理，处理完毕后，初始化缓冲区，交换缓冲区
    {
      while (1)
      {
        // 1.判断生产缓冲区有没有数据，有则交换，无则阻塞。
        // 同时要为互斥锁加一个生命周期，因为交换缓冲区并不需要对数据的处理进行加锁保护，交换完了，就释放锁
        {
          std::unique_lock<std::mutex> lock(_mutex);
          if (_stop && _pro_buf.empty())
            break;
          // 若退出前被唤醒或者有数据被唤醒，处理完。或者生产者不为空。
          _cond_con.wait(lock, [&]()
                         { return _stop || !_pro_buf.empty(); });

          _con_buf.swap(_pro_buf); // 交换缓冲区
          // 2.唤醒生产者
          if (_looper_type == AsyncType::ASYNC_SAFE)
            _cond_pro.notify_all();
        }
        // 3.唤醒后，对消费者缓冲区进行数据处理
        _callBack(_con_buf);
        // 4.初始化消费缓冲区，重置消费者缓冲区
        _con_buf.reset();
      }
    }

  private:
    AsyncType _looper_type;
    std::atomic<bool> _stop; // 工作器停止的标志
    Buffer _pro_buf;         // 生产者缓冲区
    Buffer _con_buf;         // 消费者缓冲区
    std::mutex _mutex;
    std::condition_variable _cond_pro; // 条件变量
    std::condition_variable _cond_con;
    std::thread _thread; // 异步工作器对应的工作线程
  };

}

#endif